# -*- coding: utf-8 -*-
"""
基于ALS协同过滤推荐
"""

import json
import math

import redis
from confluent_kafka import Consumer, KafkaError

from config import DevelopmentConfig
from utils.DataBase import DataBase
os.environ["PYSPARK_PYTHON"] = '/opt/module/allPythonVenv/algorithmVenv/bin/python3.7'
os.environ["PYSPARK_DRIVER_PYTHON"] = '/opt/module/allPythonVenv/algorithmVenv/bin/python3.7'
os.environ['JAVA_HOME'] = "/opt/module/jdk1.8.0_212"

class ALSRecommend:
    def __init__(self):
        self.mysql = DataBase()
        # 创建redis连接对象
        self.redis = redis.StrictRedis(
            host=DevelopmentConfig.redis_host,
            port=DevelopmentConfig.redis_port,
            decode_responses=True,  # 设置为True时，获取的数据将以字符串形式返回，而不是字节
            max_connections=10  # 设置连接池的大小
        )
        # 创建Kafka消费者对象
        self.consumer = Consumer({
            'bootstrap.servers': DevelopmentConfig.bootstrap_servers,  # Kafka服务器地址
            'group.id': DevelopmentConfig.group_id,  # 消费者组ID，确保每个消费者组有一个唯一的ID
            'auto.offset.reset': 'latest',  # 从最早的偏移量开始消费
        })
        self.sim_product_matrix = None

    def load_data(self, sql):
        # 从MySQL中获取数据
        data = self.mysql.select(sql)
        data = [(t[0], json.loads(t[1])) for t in data]
        # print(data)
        sim_product_matrix = dict()
        for row in data:
            product_score = dict()
            for term in row[1]:
                product_score[term['product_id']] = term['score']
            sim_product_matrix[row[0]] = product_score
        # print(sim_product_matrix)
        self.sim_product_matrix = sim_product_matrix

    def link_kafka(self):
        # 订阅"recommender"主题
        self.consumer.subscribe(['recommender_topic'])
        try:
            while True:
                # 从Kafka拉取消息
                msg = self.consumer.poll(10)  # 10秒超时
                if msg is None:
                    continue
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue  # 没有更多消息可供消费
                    else:
                        print(msg.error())  # 其他错误
                        break
                # 处理接收到的消息
                # noinspection PyArgumentList
                value = msg.value().decode('utf-8')  # 假设消息是UTF-8编码的字符串
                # 找到PRODUCT_RATING_PREFIX:后的数据
                prefix = "PRODUCT_RATING_PREFIX:"
                data_start = value.find(prefix) + len(prefix)
                product_data = value[data_start:]
                print(f"Received message: {product_data}")
                user_id, product_id, score, timestamp = tuple(product_data.split("|"))
                user_id, product_id, score, timestamp = int(user_id), int(product_id), float(score), int(timestamp)
                print(user_id, product_id, score, timestamp)
                # 核心算法，定义处理流程
                # ToDo: 核心算法流程
                # 1. 从redis里面取出当前用户的最近评分，保存为一个元组，[(product_id, score)]
                user_recently_ratings = self.get_user_recently_ratings(6, user_id)
                # print(user_recently_ratings)
                # 2. 从相似度矩阵中获取当前商品最相似的商品列表，作为备选列表，[product_id]
                candidate_products = self.get_top_sim_products(12, product_id, user_id)
                # 3. 计算每个备选商品的推荐优先级，得到当前用户的实时推荐列表，[(product_id, score)]
                stream_product_recs = self.compute_product_score(candidate_products, user_recently_ratings, 6, user_id)
                # 4. 把推荐列表保存到MySQL数据库
                self.save_data_to_mysql(user_id, stream_product_recs)
        except KeyboardInterrupt:
            pass
        finally:
            self.close_connection()

    def get_user_recently_ratings(self, num, user_id):
        """
        :param num: 当前用户最近的num个评分
        :param user_id: 当前用户id
        :return: [(product_id, score)]
        """
        # 从redis中用户的评分队列里获取评分数据，list键名为userid:USERID，值格式是PRODUCTID:SCORE
        result = self.redis.lrange("userId:" + str(user_id), 0, num)
        return [(int(item.split(":")[0]), float(item.split(":")[1])) for item in result]

    def get_top_sim_products(self, num, product_id, user_id):
        """
        :param num: 与当前商品最相似的num个商品
        :param product_id: 当前商品id
        :param user_id: 当前用户id
        :return: [product_id]
        """
        # 从sim_product_matrix中拿到当前商品的相似度列表
        all_sim_products_dict = self.sim_product_matrix[product_id]
        # 获取用户已经评分过的商品，过滤掉，排序输出
        rated_products = self.mysql.select(f"select product_id from ratings where user_id={user_id}")
        return sorted(
            filter(lambda key: key not in rated_products, all_sim_products_dict.keys()),
            key=lambda key: all_sim_products_dict[key],
            reverse=True
        )[:num]

    def compute_product_score(self, candidate_products, user_recently_ratings, num, user_id):
        """
        :param candidate_products: 备选商品：[product_id]
        :param user_recently_ratings: 最近评分：[(product_id, score)]
        :param num: 备选商品中推荐得分的前num个
        :param user_id: 当前用户id
        :return: 计算每个备选商品的推荐得分：[(product_id, score)]
        """
        # 定义结果字典，用于保存备选商品得分{product_id:score}
        recommendations = dict()

        # 遍历每个备选商品，计算和已评分商品的相似度
        for candidate_product_id in candidate_products:
            # 从商品相似度矩阵中获取当前备选商品和当前已评分商品间的相似度
            base_score = []
            incre_count = 0
            decre_count = 0
            for rated_product_id, rated_score in user_recently_ratings:
                sim_score = self.sim_product_matrix.get(candidate_product_id, None).get(rated_product_id, 0)
                if sim_score > 0.4:
                    # 按照公式计算推荐得分
                    base_score.append(sim_score * rated_score)
                    # 统计奖励项和惩罚项
                    if rated_score > 3:
                        incre_count += 1
                    else:
                        decre_count += 1
            if len(base_score) >= 1:
                recommendations[candidate_product_id] = (math.fsum(base_score) / len(base_score)  # 基础分
                                                         + math.log(max(incre_count, 1))  # 奖励项
                                                         + math.log(max(decre_count, 1)))  # 惩罚项
            else:
                recommendations[candidate_product_id] = 0

        # 上一次给的推荐
        sql = f"select recommendations from stream_user_recs where user_id={user_id}"
        if self.mysql.select(sql) != ():
            previous_recommendations = json.loads(self.mysql.select(sql)[0][0])
            previous_recommendations = [(term['product_id'], term['score']) for term in previous_recommendations]
        else:
            previous_recommendations = []

        current_recommendations = list(set(recommendations.items()) | set(previous_recommendations))  # 求并集
        return sorted(current_recommendations, key=lambda x: x[1], reverse=True)[:num]

    def save_data_to_mysql(self, user_id, stream_product_recs):
        recommendations = json.dumps(
            [{"product_id": product_id, "score": score} for product_id, score in stream_product_recs]
        )
        data = (recommendations, user_id)
        # print(data)
        # 检查user_id是否存在
        check_query = f"SELECT * FROM stream_user_recs WHERE user_id = {user_id}"
        result = self.mysql.select(check_query)
        # 如果存在则更新，否则插入新记录
        if result:
            update_query = "UPDATE stream_user_recs SET recommendations = %s WHERE user_id = %s"
            self.mysql.sql_submit(update_query, data)
        else:
            insert_query = "INSERT INTO stream_user_recs (recommendations, user_id) VALUES (%s, %s)"
            self.mysql.sql_submit(insert_query, data)

    def close_connection(self):
        self.redis.close()
        self.consumer.close()


if __name__ == '__main__':
    als_Recommender = ALSRecommend()
    # 加载MySQL数据，商品相似度矩阵
    als_Recommender.load_data("select product_id,recommendations from als_product_recs;")
    # 连接Kafka，开始实时处理
    als_Recommender.link_kafka()
